package com.zyz.zchat;

import java.io.IOException;
import java.io.OutputStream;
import java.net.*;
import java.util.HashMap;
import java.util.HashSet;
import java.util.concurrent.*;

public class BroadcastManager {
    HashSet<DatagramSocket> listeners;
    ThreadPoolExecutor listenerPool;
    HashSet<DatagramSocket> senders;
    ThreadPoolExecutor senderPool;
    HashSet<InetAddress> hosts;
    InetAddress BroadcastAddr;
    final String flag="ZChat";
    DatagramPacket flagPacket;
    public static HashSet<Integer>acceptedPorts;
    static {
        acceptedPorts=new HashSet<>();
        acceptedPorts.add(2001);
        acceptedPorts.add(414);

    }
    private static final int CORE_POOL_SIZE = 5;
    private static final int MAX_POOL_SIZE = 10;
    private static final int QUEUE_CAPACITY = 100;
    private static final Long KEEP_ALIVE_TIME = 1L;
    void broadcast() throws IOException {
        for (DatagramSocket socket:senders)
            senderPool.execute(() -> {
                    try {
                        flagPacket.setPort(socket.getPort());
                        flagPacket.setAddress(socket.getInetAddress());
                        socket.send(flagPacket);
                    } catch (IOException e) {
                        e.printStackTrace();
                        System.out.println(socket.getPort());
                    }
            });
    }
    void start()
    {
        for (DatagramSocket s:listeners) {
            listenerPool.execute(() -> listen(s));
        }
    }

    void listen(DatagramSocket s)
    {
        DatagramPacket packet;
        byte[] buf=new byte[1024];
        packet=new DatagramPacket(buf,buf.length);
        while(!s.isClosed())
        {
            try {
                s.receive(packet);
                buf=packet.getData();
                String data=(new String(buf)).trim();
                if( data.trim().equals(flag))
                {
                    hosts.add(packet.getAddress());
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    public BroadcastManager()
    {
        try {
            BroadcastAddr=InetAddress.getByName("255.255.255.255");
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        flagPacket=new DatagramPacket(flag.getBytes(),flag.getBytes().length);
        listeners=new HashSet<>();
        hosts=new HashSet<>();
        senders=new HashSet<>();
        for (Integer i:acceptedPorts) {
            try {
                DatagramSocket listener=new DatagramSocket(i,InetAddress.getLocalHost());
                DatagramSocket sender=new DatagramSocket();
                sender.connect(BroadcastAddr,i);
                listeners.add(listener);
                senders.add(sender);
            } catch (IOException e) {
                System.out.println("port:"+i+" is not available");
                e.printStackTrace();
            }
        }
        listenerPool= new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
        senderPool= new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_POOL_SIZE,
                KEEP_ALIVE_TIME,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(QUEUE_CAPACITY),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
        start();
    }

    public static void main(String[] args) {
        BroadcastManager m=new BroadcastManager();
        m.start();
        try {
            m.broadcast();

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
